package org.filesort.impl;

import java.io.*;
import java.lang.ref.SoftReference;
import java.util.Arrays;

import javax.annotation.concurrent.NotThreadSafe;

import com.google.common.annotations.VisibleForTesting;
import org.filesort.IFileContentProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.google.common.base.Preconditions.checkArgument;

/**
 * Implementation {@link IFileContentProcessor} for file of longs (64 bites)
 *
 * @author cheremin
 * @since 10.02.12,  16:58
 */
@NotThreadSafe
public class LongsProcessor implements IFileContentProcessor {
    private static final Logger log = LoggerFactory.getLogger( LongsProcessor.class );

    static final int LONG_SIZE = Long.SIZE / Byte.SIZE;

    private static final int DEFAULT_BUFFER_SIZE = 1 << 18;//256Kb

    private final transient byte[] buffer;


    public LongsProcessor() {
        this( DEFAULT_BUFFER_SIZE );
    }

    public LongsProcessor( final int bufferSize ) {
        checkArgument( bufferSize > 0, "bufferSize(%s) must be >0", bufferSize );
        checkArgument( bufferSize % dataElementSize() == 0, "bufferSize(%s) % dataSize(%s) != 0", bufferSize, dataElementSize() );

        buffer = new byte[bufferSize];
    }

    @Override
    public int dataElementSize() {
        return LONG_SIZE;
    }

    @Override
    public void merge( final IBytesSource[] sources,
                       final long maxRunLength,
                       final IBytesTarget target ) throws IOException {
        final SourceReadBuffer[] sourceBuffers = new SourceReadBuffer[sources.length];
        final int longReadBufferSize = buffer.length / dataElementSize();
        for ( int i = 0; i < sources.length; i++ ) {
            final IBytesSource source = sources[i];
            sourceBuffers[i] = new SourceReadBuffer( source, longReadBufferSize );
        }
        final TargetWriteBuffer targetBuffer = new TargetWriteBuffer(
                target,
                longReadBufferSize * sources.length
        );
        while ( true ) {
            long minValue = Long.MAX_VALUE;
            int minValueSourceIndex = -1;
            for ( int i = 0; i < sourceBuffers.length; i++ ) {
                final SourceReadBuffer sourceBuffer = sourceBuffers[i];
                if ( sourceBuffer.hasElement() ) {
                    final long l = sourceBuffer.peek();
                    if ( l <= minValue ) {
                        minValue = l;
                        minValueSourceIndex = i;
                    }
                }
            }
            if ( minValueSourceIndex == -1 ) {
                break;//all sources are exhausted
            }
            targetBuffer.put( minValue );

            sourceBuffers[minValueSourceIndex].take();
        }
        targetBuffer.flush();
    }

    private class SourceReadBuffer {
        private final IBytesSource source;
        private final long[] buffer;
        private int bufferPosition;
        private int bufferLength;

        private SourceReadBuffer( final IBytesSource source,
                                  final int bufferSize ) {
            checkArgument( bufferSize > 0, "bufferSize(%s) must be > 0", bufferSize );
            this.source = source;
            this.buffer = new long[bufferSize];
            bufferPosition = -1;
            bufferLength = -1;
        }

        public long peek() throws IOException {
            if ( hasElement() ) {
                return buffer[bufferPosition];
            } else {
                throw new EOFException( "EOS reached" );
            }
        }

        public long take() throws IOException {
            if ( hasElement() ) {
                final long l = buffer[bufferPosition];
                bufferPosition++;
                return l;
            } else {
                throw new EOFException( "EOS reached" );
            }
        }

        public boolean hasElement() throws IOException {
            ensureBufferFilled();
            return bufferPosition < bufferLength;
        }

        private void ensureBufferFilled() throws IOException {
            if ( bufferLength == 0 ) {
                return;//== end-of-source reached
            }
            if ( bufferPosition == bufferLength ) {
                bufferLength = readArrayNonStrict(
                        source,
                        buffer,
                        buffer.length
                );
                bufferPosition = 0;
            }
        }


    }

    private class TargetWriteBuffer {
        private final IBytesTarget target;
        private final long[] buffer;
        private int bufferPosition;

        private TargetWriteBuffer( final IBytesTarget target,
                                   final int bufferSize ) {
            checkArgument( bufferSize > 0, "bufferSize(%s) must be > 0", bufferSize );
            this.target = target;
            this.buffer = new long[bufferSize];
            bufferPosition = 0;
        }

        public void put( final long value ) throws IOException {
            ensureBufferSpaceAvailable();
            buffer[bufferPosition] = value;
            bufferPosition++;
        }

        private void ensureBufferSpaceAvailable() throws IOException {
            if ( bufferPosition == buffer.length ) {
                flush();
            }
        }

        public void flush() throws IOException {
            if ( bufferPosition > 0 ) {
                writeArray( target, buffer, bufferPosition );
                bufferPosition = 0;
            }
        }
    }

    @Override
    public void sortRun( final IBytesSource source,
                         final IBytesTarget target,
                         final int runLength ) throws IOException {
        checkArgument( runLength % dataElementSize() == 0, "runLength=%s", runLength );

        final int longsCount = runLength / dataElementSize();

        final long[] array = allocate( longsCount );

        readArrayStrict( source, array, longsCount );

        Arrays.sort( array, 0, longsCount );

        writeArray( target, array, longsCount );
    }

    @VisibleForTesting
    void writeArray( final IBytesTarget target,
                     final long[] array,
                     final int longsCount ) throws IOException {
        final int longsInBuffer = buffer.length / LONG_SIZE;

        int longsIndex = 0;
        while ( longsIndex < longsCount ) {
            final int longsLeft = longsCount - longsIndex;
            final int longsToCopy = Math.min(
                    longsInBuffer,
                    longsLeft
            );

            longsToBytes( array, longsIndex, longsToCopy, buffer, 0 );

            final int bytesCopied = longsToCopy * LONG_SIZE;

            target.put( buffer, bytesCopied );

            longsIndex += longsToCopy;
        }
    }

    @VisibleForTesting
    void readArrayStrict( final IBytesSource source,
                          final long[] array,
                          final int longsCount ) throws IOException {
        final int actuallyRead = readArrayNonStrict( source, array, longsCount );
        if ( actuallyRead < longsCount ) {
            throw new EOFException(
                    "bytesRead=0 (=> end-of-source) but not all requested data" +
                            " was read (actuallyRead=" + actuallyRead + ", longsCount=" + longsCount + ")"
            );
        }
    }

    @VisibleForTesting
    int readArrayNonStrict( final IBytesSource source,
                            final long[] array,
                            final int longsCount ) throws IOException {
        int longsIndex = 0;
        while ( longsIndex < longsCount ) {
            final int bytesRead = source.get( buffer );

            assert ( bytesRead % dataElementSize() == 0 ) : "bytesRead=" + bytesRead;

            if ( bytesRead == 0 ) {
                break;
            }

            final int longsRead = bytesRead / LONG_SIZE;
            bytesToLongs( buffer, 0, bytesRead, array, longsIndex );

            longsIndex += longsRead;
        }
        return longsIndex;
    }

    private transient SoftReference<long[]> cachedArrayRef = new SoftReference<long[]>( null );

    private long[] allocate( final int length ) {
        long[] array = cachedArrayRef.get();
        if ( array == null || array.length < length ) {
            array = new long[length];
            cachedArrayRef = new SoftReference<long[]>( array );
        }
        return array;
    }


    public static void longsToBytes( final long[] longsArray,
                                     final int longsOffset,
                                     final int longsCount,
                                     final byte[] bytesArray,
                                     final int byteOffset ) {
        checkArgument( longsCount * LONG_SIZE <= bytesArray.length - byteOffset,
                       "longsCount=%s, bytesFree=%s",
                       longsCount, bytesArray.length - byteOffset
        );

        int bytesIndex = byteOffset;
        for ( int i = 0; i < longsCount; i++ ) {
            final long l = longsArray[i + longsOffset];
            bytesArray[bytesIndex + 0] = ( byte ) ( l >>> 56 );
            bytesArray[bytesIndex + 1] = ( byte ) ( l >>> 48 );
            bytesArray[bytesIndex + 2] = ( byte ) ( l >>> 40 );
            bytesArray[bytesIndex + 3] = ( byte ) ( l >>> 32 );
            bytesArray[bytesIndex + 4] = ( byte ) ( l >>> 24 );
            bytesArray[bytesIndex + 5] = ( byte ) ( l >>> 16 );
            bytesArray[bytesIndex + 6] = ( byte ) ( l >>> 8 );
            bytesArray[bytesIndex + 7] = ( byte ) ( l >>> 0 );

            bytesIndex += LONG_SIZE;
        }
    }

    public static void bytesToLongs( final byte[] bytesArray,
                                     final int byteOffset,
                                     final int byteCount,
                                     final long[] longsArray,
                                     final int longsOffset ) {
        checkArgument( byteCount <= bytesArray.length - byteOffset,
                       "byteCount(%s) must be < bytesArray.length-byteOffset (%s)",
                       byteCount, bytesArray.length - byteOffset
        );
        final int longsCount = byteCount / LONG_SIZE;

        for ( int i = 0; i < longsCount; i++ ) {
            final int bufferIndex = i * LONG_SIZE + byteOffset;
            longsArray[longsOffset + i] = makeLong(
                    bytesArray[bufferIndex + 7],
                    bytesArray[bufferIndex + 6],
                    bytesArray[bufferIndex + 5],
                    bytesArray[bufferIndex + 4],
                    bytesArray[bufferIndex + 3],
                    bytesArray[bufferIndex + 2],
                    bytesArray[bufferIndex + 1],
                    bytesArray[bufferIndex]
            );
        }
    }

    private static long makeLong( final byte b7,
                                  final byte b6,
                                  final byte b5,
                                  final byte b4,
                                  final byte b3,
                                  final byte b2,
                                  final byte b1,
                                  final byte b0 ) {
        return ( ( ( ( long ) b0 & 0xff ) << 56 ) |
                ( ( ( long ) b1 & 0xff ) << 48 ) |
                ( ( ( long ) b2 & 0xff ) << 40 ) |
                ( ( ( long ) b3 & 0xff ) << 32 ) |
                ( ( ( long ) b4 & 0xff ) << 24 ) |
                ( ( ( long ) b5 & 0xff ) << 16 ) |
                ( ( ( long ) b6 & 0xff ) << 8 ) |
                ( ( ( long ) b7 & 0xff ) << 0 ) );
    }
}
